package com.apes.framework.plugin.mq.store;

import com.apes.framework.api.SpringManager;
import com.apes.framework.config.elasticsearch.ApesElasticsearchHealthIndicator;
import com.apes.framework.plugin.esb.api.SimpleRequest;
import com.apes.framework.plugin.mq.Consumer;
import com.apes.framework.plugin.mq.MessageChannel;
import com.apes.framework.plugin.mq.domain.MQueue;
import com.apes.framework.plugin.mq.domain.MQueueBak;
import com.apes.framework.plugin.mq.domain.MQueueDeadLetter;
import com.apes.framework.plugin.mq.log.MQLog;
import com.apes.framework.plugin.mq.log.MQLogRepository;
import com.apes.framework.util.DateUtil;
import com.apes.framework.util.MD5Util;
import com.apes.framework.util.MapUtil;
import com.apes.framework.util.Tools;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.actuate.health.Status;
import org.springframework.core.env.Environment;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Sort;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.messaging.MessageHeaders;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import javax.annotation.Resource;
import java.util.*;


/**
 * 功能：消息存储服务
 *
 * @author xul
 * @create 2017-12-07 16:34
 */
@Component("mqStoreService")
public class MQStoreService {

    @Autowired
    private Environment env;

    @Autowired
    private MongoTemplate template;

    @Autowired
    private QueueRepository queueRepository;

    @Autowired
    private QueueBakRepository queueBakRepository;

    @Autowired
    private QueueDeadLetterRepository deadLetterRepository;

    @Autowired
    private MQLogRepository logRepository;

    private ApesElasticsearchHealthIndicator healthIndicator;

    public MQStoreService(ApesElasticsearchHealthIndicator healthIndicator) {
        this.healthIndicator = healthIndicator;
    }

    public MQueue save(MQueue message) {
        return queueRepository.save(message);
    }

    public MQueue findOne(String id) {
        return queueRepository.findById(id).orElse(null);
    }

    public void delete(MQueue message) {
        queueRepository.delete(message);
    }

    public void backup(MQueue message) {
        MQueueBak mqb = new MQueueBak();
        Tools.beanConvert(mqb, message);
        mqb.setState(2);
        queueBakRepository.save(mqb);
    }

    public void save(MQueueDeadLetter message) {
        deadLetterRepository.save(message);
    }

    public void delete(MQueueDeadLetter message) {
        deadLetterRepository.delete(message);
    }

    public List<String> getTopics() {
        List<String> topics = new ArrayList<>();
        template.getCollection("MQueue").distinct("topic", String.class).into(topics);
        return topics;
    }

    private int getSize() {
        String size = env.getProperty("spring.mq.pageSize");
        return Integer.valueOf(StringUtils.isEmpty(size) ? "5000" : size);
    }

    /**
     * 获取队列消息
     *
     * @param request
     * @return
     */
    public Page<MQueue> findByTopic(SimpleRequest request) {
        List<Map> condition = request.get("condition");
        Optional<Object> value = condition.stream().map(s -> s.get("value")).findFirst();
        PageRequest pageRequest = PageRequest.of(request.get("page"), request.get("size"));
        if (value.isPresent()) {
            return queueRepository.findByTopicLikeOrderByEndedDesc((String) value.get(), pageRequest);
        }
        return queueRepository.findAll(pageRequest);
    }

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    @Value("${rocketmq.tags:}")
    private String tags;

    /**
     * RocketMQ上线消息初始化
     */
    public void init() {
        int size = getSize();
        String begin = null;
        String end = DateUtil.format(new Date(), DateUtil.SHOW_DATETIME_MILLI_FORMAT);
        while (true) {
            Criteria criteria = Criteria.where("state").is(1).and("ended").lte(end);
            List<MQueue> messages = getMessages(criteria, begin, size);
            messages.stream().forEach(message -> {
                String rocketMQTopic = SpringManager.getBean(MessageChannel.class).getRocketMQTopic(message.getTopic(), message.getTags());
                if (StringUtils.isEmpty(rocketMQTopic)) {
                    return;
                }
                if (StringUtils.hasText(tags) && StringUtils.isEmpty(message.getTags())) {
                    message.setTags(tags);
                }
                message.setTimes(0);
                message = queueRepository.save(message);
                Map<String, Object> headers = MapUtil.mapper(RocketMQHeaders.KEYS, message.getId());
                rocketMQTemplate.convertAndSend(rocketMQTopic, message.getId(), new MessageHeaders(headers));
            });
            if (messages.size() < size) {
                break;
            }
            begin = messages.get(messages.size() - 1).getCreated();
        }
    }

    private List<MQueue> getMessages(Criteria criteria, String begin, int size) {
        if (StringUtils.hasText(begin)) {
            criteria.and("created").gt(begin);
        }
        Query query = new Query(criteria).with(PageRequest.of(0, size)).with(new Sort(Sort.Direction.ASC, "created"));
        return template.find(query, MQueue.class);
    }

    /**
     * 定时任务：处理遗漏消息
     **/
    public void handlingMissingMessages() {
        int size = getSize();
        String begin = null;
        /*单位：分钟**/
        String end = DateUtil.format(DateUtil.add(new Date(), Calendar.MINUTE, -1 * 10), DateUtil.SHOW_DATETIME_MILLI_FORMAT);
        while (true) {
            Criteria criteria = Criteria.where("state").is(0).and("handled").is(0).and("ended").lte(end);
            List<MQueue> messages = getMessages(criteria, begin, size);
            messages.forEach(message -> {
                String rocketMQTopic = SpringManager.getBean(MessageChannel.class).getRocketMQTopic(message.getTopic(), message.getTags());
                if (StringUtils.hasText(rocketMQTopic)) {
                    message.setHandled(1);
                    queueRepository.save(message);
                    Map<String, Object> headers = MapUtil.mapper(RocketMQHeaders.KEYS, message.getId());
                    rocketMQTemplate.convertAndSend(rocketMQTopic, message.getId(), new MessageHeaders(headers));
                }
            });
            if (messages.size() < size) {
                break;
            }
            begin = messages.get(messages.size() - 1).getCreated();
        }
    }

    private Map<String, Consumer> getConsumers() {
        return SpringManager.getBean(MessageChannel.class).getConsumers();
    }

    /**
     * 定时任务：处理死信消息
     **/
    public void handlingDeadLetter() {
        int size = getSize();
        /*单位：小时**/
        String end = DateUtil.format(DateUtil.add(new Date(), Calendar.HOUR_OF_DAY, -1 * 24), DateUtil.SHOW_DATETIME_MILLI_FORMAT);
        while (true) {
            Criteria criteria = Criteria.where("created").lte(end);
            Query query = new Query(criteria).with(PageRequest.of(0, size)).with(new Sort(Sort.Direction.ASC, "created"));
            List<MQueueDeadLetter> deadLetters = template.find(query, MQueueDeadLetter.class);
            handle(deadLetters);
            if (deadLetters.size() < size) {
                break;
            }
        }
    }

    private void handle(List<MQueueDeadLetter> deadLetters) {
        deadLetters.forEach(deadLetter -> {
            MQueue message = queueRepository.findById(deadLetter.getMsgId()).orElse(null);
            if (message != null) {
                String rocketMQTopic = SpringManager.getBean(MessageChannel.class).getRocketMQTopic(message.getTopic(), message.getTags());
                if (StringUtils.hasText(rocketMQTopic)) {
                    message.setTimes(0);
                    message = queueRepository.save(message);
                    Map<String, Object> headers = MapUtil.mapper(RocketMQHeaders.KEYS, message.getId());
                    rocketMQTemplate.convertAndSend(rocketMQTopic, message.getId(), new MessageHeaders(headers));
                }
            }
            delete(deadLetter);
        });
    }

    /***
     * 手动执行死信消息
     */
    public void handlingDeadLetterById(SimpleRequest request) {
        Criteria criteria = Criteria.where("msgId").is(request.get("id"));
        Query query = new Query(criteria);
        List<MQueueDeadLetter> deadLetters = template.find(query, MQueueDeadLetter.class);
        handle(deadLetters);
    }

    public MQLog createLog(MQueue message) {
        if (message == null) return null;
        MQLog log = new MQLog();
        log.setCreatedOnTimeStamp(System.currentTimeMillis());
        log.setMessageId(message.getId());
        log.setTopic(message.getTopic());
        log.setTags(message.getTags());
        log.setBody(message.getBody());
        log.setId(MD5Util.encrypt(log.getMessageId() + log.getCreatedOn()));
        boolean health = healthIndicator != null ? healthIndicator.health().getStatus() == Status.UP : false;
        if (!health) return null;
        return logRepository.save(log);
    }

    public void updateLog(MQLog log) {
        if (log == null) return;
        log.setEndedOn(DateUtil.format("yyyy-MM-dd'T'HH:mm:ss.SSS") + "+0800");
        log.setEndOnTimeStamp(System.currentTimeMillis());
        log.setTime(log.getEndOnTimeStamp() - log.getCreatedOnTimeStamp());
        logRepository.save(log);
    }
    public Map MQTotal (SimpleRequest request) {
        String topic = request.getJO().getString("topic");
        MQueue message = queueRepository.findById(topic).orElse(null);
        PageRequest pageRequest = PageRequest.of(0, 20);
        Page<MQueue> byTopicLikeOrderByEndedDesc = queueRepository.findByTopicLikeAndStateOrderByEndedDesc(topic, 0,pageRequest);
        return MapUtil.mapper("total", byTopicLikeOrderByEndedDesc.getTotalElements());
    }
}
